package com.carrot.mq.mqserver.datacenter;

import com.carrot.mq.common.MqException;
import com.carrot.mq.mqserver.core.Binding;
import com.carrot.mq.mqserver.core.Exchange;
import com.carrot.mq.mqserver.core.MSGQueue;
import com.carrot.mq.mqserver.core.Message;

import java.io.IOException;
import java.util.LinkedList;
import java.util.List;
import java.util.concurrent.ConcurrentHashMap;

/**
 * @author 兴趣使然黄小黄
 * @version 1.0
 * @date 2023/8/11 22:03
 * 统一管理内存上的数据
 */
public class MemoryDataCenter {
    // <exchangeName, exchange>
    private ConcurrentHashMap<String, Exchange> exchangeMap = new ConcurrentHashMap<>();
    // <queueName, MSGQueue>
    private ConcurrentHashMap<String, MSGQueue> queueMap = new ConcurrentHashMap<>();
    // <exchangeName, <queueName, binding>>
    private ConcurrentHashMap<String, ConcurrentHashMap<String, Binding>> bindingsMap = new ConcurrentHashMap<>();
    // <messageId, message>
    private ConcurrentHashMap<String, Message> messageMap = new ConcurrentHashMap<>();
    // <queueName, list<message>>
    private ConcurrentHashMap<String, LinkedList<Message>> queueMessageMap = new ConcurrentHashMap<>();
    // <queueName, <messageId, message>>
    private ConcurrentHashMap<String, ConcurrentHashMap<String, Message>> queueMessageWaitAckMap = new ConcurrentHashMap<>();

    /**
     * 添加交换机
     */
    public void insertExchange(Exchange exchange) {
        exchangeMap.put(exchange.getName(), exchange);
        System.out.println("[MemoryDataCenter] 交换机添加成功! exchangeName=" + exchange.getName());
    }

    /**
     * 获取交换机
     */
    public Exchange getExchange(String exchangeName) {
        return exchangeMap.get(exchangeName);
    }

    /**
     * 删除交换机
     */
    public void deleteExchange(String exchangeName) {
        exchangeMap.remove(exchangeName);
        System.out.println("[MemoryDataCenter] 交换机删除成功! exchangeName=" + exchangeName);
    }

    /**
     * 添加队列
     */
    public void insertQueue(MSGQueue queue) {
        queueMap.put(queue.getName(), queue);
        System.out.println("[MemoryDataCenter] 队列添加成功! queueName=" + queue.getName());
    }

    /**
     * 获取队列
     */
    public MSGQueue getQueue(String queueName) {
        return queueMap.get(queueName);
    }

    /**
     * 删除队列
     */
    public void deleteQueue(String queueName) {
        queueMap.remove(queueName);
        System.out.println("[MemoryDataCenter] 队列删除成功! queueName=" + queueName);
    }

    /**
     * 添加绑定
     */
    public void insertBinding(Binding binding) throws MqException {
        // 查询 exchangeName 对应的 Map 是否存在, 不存在则新建一个 Map 存储
//        ConcurrentHashMap<String, Binding> bindingMap = bindingsMap.get(binding.getExchangeName());
//        if (bindingMap == null) {
//            bindingMap = new ConcurrentHashMap<>();
//            bindingsMap.put(binding.getExchangeName(), bindingMap);
//        }
        ConcurrentHashMap<String, Binding> bindingMap = bindingsMap.computeIfAbsent(binding.getExchangeName(),
                k -> new ConcurrentHashMap<>());
        synchronized (bindingMap) {
            // 根据 queueName 查询对应的 binding 是否存在, 如果存在则抛出异常
            if (bindingMap.get(binding.getQueueName()) != null) {
                throw new MqException("[MemoryDataCenter] 绑定已经存在! exchangeName=" + binding.getExchangeName() +
                        ", queueName=" + binding.getQueueName());
            }
            // 插入操作
            bindingMap.put(binding.getQueueName(), binding);
        }
        System.out.println("[MemoryDataCenter] 绑定添加成功! exchangeName=" + binding.getExchangeName() +
                ", queueName=" + binding.getQueueName());
    }

    /**
     * 通过 exchangeName 和 queueName 获取唯一的绑定
     */
    public Binding getBinding(String exchangeName, String queueName) {
        ConcurrentHashMap<String, Binding> bindingMap = bindingsMap.get(exchangeName);
        if (bindingMap == null) {
            return null;
        }
        return bindingMap.get(queueName);
    }

    /**
     * 通过 exchangeName 获取该交换机上的所有绑定
     */
    public ConcurrentHashMap<String, Binding> getBindings(String exchangeName) {
        return bindingsMap.get(exchangeName);
    }

    /**
     * 删除绑定
     */
    public void deleteBinding(Binding binding) throws MqException {
        ConcurrentHashMap<String, Binding> bindingMap = bindingsMap.get(binding.getExchangeName());
        if (bindingMap == null) {
            // 该交换机上没有绑定任何队列
            throw new MqException("[MemoryDataCenter] 绑定不存在! exchangeName=" + binding.getExchangeName() +
                    ", queueName=" + binding.getQueueName());
        }
        bindingMap.remove(binding.getQueueName());
        System.out.println("[MemoryDataCenter] 绑定删除成功! exchangeName=" + binding.getExchangeName() +
                ", queueName=" + binding.getQueueName());
    }

    /**
     * 添加消息
     */
    public void addMessage(Message message) {
        messageMap.put(message.getMessageId(), message);
        System.out.println("[MemoryDataCenter] 新消息添加成功! messageId=" + message.getMessageId());
    }

    /**
     * 根据 id 查询消息
     */
    public Message getMessage(String messageId) {
        return messageMap.get(messageId);
    }

    /**
     * 根据 id 删除消息
     */
    public void removeMessage(String messageId) {
        messageMap.remove(messageId);
        System.out.println("[MemoryDataCenter] 消息被移除! messageId=" + messageId);
    }

    /**
     * 发送消息到指定队列
     */
    public void sendMessage(MSGQueue queue, Message message) {
        // 将消息放到对应的队列数据结构中
        // 根据队列名找到对应的消息链表
        LinkedList<Message> messages = queueMessageMap.computeIfAbsent(queue.getName(), k -> new LinkedList<>());
        // 将数据添加到 messages 内
        synchronized (messages) {
            messages.add(message);
        }
        // 消息中心插入消息
        addMessage(message);
        System.out.println("[MemoryDataCenter] 消息被投递到队列中! messageId=" + message.getMessageId());
    }

    /**
     * 从指定队列取消息
     */
    public Message pollMessage(String queueName) {
        // 根据队列名查找对应的消息链表
        LinkedList<Message> messages = queueMessageMap.get(queueName);
        if (messages == null) {
            return null;
        }
        synchronized (messages) {
            if (messages.size() == 0) {
                return null;
            }
            // 链表中有元素, 则进行头删
            Message currentMessage = messages.remove(0);
            System.out.println("[MemoryDataCenter] 消息从队列中取出! messageId=" + currentMessage.getMessageId());
            return currentMessage;
        }
    }

    /**
     * 获取指定队列中的消息个数
     */
    public int getMessageCount(String queueName) {
        LinkedList<Message> messages = queueMessageMap.get(queueName);
        if (messages == null) {
            return 0;
        }
        synchronized (messages) {
            return messages.size();
        }
    }

    /**
     * 添加未确认消息
     */
    public void addMessageWaitAck(String queueName, Message message) {
        ConcurrentHashMap<String, Message> messageHashMap = queueMessageWaitAckMap.computeIfAbsent(queueName,
                k -> new ConcurrentHashMap<>());
        messageHashMap.put(message.getMessageId(), message);
        System.out.println("[MemoryDataCenter] 消息进入待确认队列! messageId=" + message.getMessageId());
    }

    /**
     * 删除未确认的消息, 当消息已经确认时调用
     */
    public void removeMessageWaitAck(String queueName, String messageId) {
        ConcurrentHashMap<String, Message> messageHashMap = queueMessageWaitAckMap.get(queueName);
        if (messageHashMap == null) {
            return;
        }
        messageHashMap.remove(messageId);
        System.out.println("[MemoryDataCenter] 消息从待确认队列删除! messageId=" + messageId);
    }

    /**
     * 获取指定的未确认消息
     */
    public Message getMessageWaitAck(String queueName, String messageId) {
        ConcurrentHashMap<String, Message> messageHashMap = queueMessageWaitAckMap.get(queueName);
        if (messageHashMap == null) {
            return null;
        }
        return messageHashMap.get(messageId);
    }

    /**
     * 从硬盘上读取队列, 将硬盘上之前持久化存储的数据恢复到内存中
     */
    public void recovery(DiskDataCenter diskDataCenter) throws IOException, MqException, ClassNotFoundException {
        // 清空之前的数据
        exchangeMap.clear();
        queueMap.clear();
        bindingsMap.clear();
        messageMap.clear();
        queueMessageMap.clear();
        // 恢复交换机数据
        List<Exchange> exchanges = diskDataCenter.selectAllExchange();
        for (Exchange exchange : exchanges) {
            exchangeMap.put(exchange.getName(), exchange);
        }
        // 恢复队列数据
        List<MSGQueue> queues = diskDataCenter.selectAllQueues();
        for (MSGQueue queue : queues) {
            queueMap.put(queue.getName(), queue);
        }
        // 恢复所有的绑定数据
        List<Binding> bindings = diskDataCenter.selectAllBindings();
        for (Binding binding : bindings) {
            ConcurrentHashMap<String, Binding> bindingMap = bindingsMap.computeIfAbsent(binding.getExchangeName(),
                    k -> new ConcurrentHashMap<>());
            bindingMap.put(binding.getQueueName(), binding);
        }
        // 恢复所有的消息数据
        // 遍历所有的队列, 根据每个队列的名字获取所有的消息
        for (MSGQueue queue : queues) {
            LinkedList<Message> messages = diskDataCenter.loadAllMessageFromQueue(queue.getName());
            queueMessageMap.put(queue.getName(), messages);
            for (Message message : messages) {
                messageMap.put(message.getMessageId(), message);
            }
        }
        // 针对 "未确认的消息" 这部分的内存数据, 不需要从硬盘中恢复
        // 等待 ack 的过程中, 服务器重启, 此时这些 "未被确认的消息" 会恢复成 "未被取走的消息", 即在硬盘上存储的消息
    }
}
